from typing import Any, Dict, Generator, Tuple

import numpy as np
import torch
import torch.nn as nn
from scipy.signal import lfilter
from torch.optim import Adam
from torch.utils.tensorboard import SummaryWriter

from ....exp.logger import Logger
from ....network.utils import get_available_device, weight_initialize
from .agent import ActorCritic


class PPOBuffer:
    """
    A buffer for storing trajectories experienced by a PPO agent interacting
    with the environment, and using Generalized Advantage Estimation (GAE-Lambda)
    for calculating the advantages of state-action pairs.
    """

    def __init__(self, size, gamma=0.99, lam=0.95):
        self.obs_buf = np.empty(size, dtype=object)
        self.act_buf = np.empty(size, dtype=object)
        self.adv_buf = np.empty(size, dtype=np.float32)
        self.rew_buf = np.empty(size, dtype=np.float32)
        self.ret_buf = np.empty(size, dtype=np.float32)
        self.val_buf = np.empty(size, dtype=np.float32)
        self.logp_buf = np.empty(size, dtype=np.float32)
        self.gamma, self.lam = gamma, lam
        self.ptr, self.path_start_idx, self.max_size = 0, 0, size

    def store(self, obs, act, rew, val, logp):
        """
        Append one timestep of agent-environment interaction to the buffer.
        """
        assert self.ptr < self.max_size  # buffer has to have room so you can store
        self.obs_buf[self.ptr] = obs
        self.act_buf[self.ptr] = act
        self.rew_buf[self.ptr] = rew
        self.val_buf[self.ptr] = val
        self.logp_buf[self.ptr] = logp
        self.ptr += 1

    def finish_path(self, last_val=0):
        """
        Call this at the end of a trajectory, or when one gets cut off
        by an epoch ending. This looks back in the buffer to where the
        trajectory started, and uses rewards and value estimates from
        the whole trajectory to compute advantage estimates with GAE-Lambda,
        as well as compute the rewards-to-go for each state, to use as
        the targets for the value function.

        The "last_val" argument should be 0 if the trajectory ended
        because the agent reached a terminal state (died), and otherwise
        should be V(s_T), the value function estimated for the last state.
        This allows us to bootstrap the reward-to-go calculation to account
        for timesteps beyond the arbitrary episode horizon (or epoch cutoff).
        """

        path_slice = slice(self.path_start_idx, self.ptr)
        rews = np.append(self.rew_buf[path_slice], last_val)
        vals = np.append(self.val_buf[path_slice], last_val)

        # the next two lines implement GAE-Lambda advantage calculation
        deltas = rews[:-1] + self.gamma * vals[1:] - vals[:-1]
        self.adv_buf[path_slice] = discount_cumsum(deltas, self.gamma * self.lam)

        # the next line computes rewards-to-go, to be targets for the value function
        self.ret_buf[path_slice] = discount_cumsum(rews, self.gamma)[:-1]

        self.path_start_idx = self.ptr

    def get(self):
        """
        Call this at the end of an epoch to get all of the data from
        the buffer, with advantages appropriately normalized (shifted to have
        mean zero and std one). Also, resets some pointers in the buffer.
        """
        # assert self.ptr == self.max_size  # buffer has to be full before you can get
        self.ptr, self.path_start_idx = 0, 0
        # the next two lines implement the advantage normalization trick
        data = dict(
            obs=self.obs_buf,
            act=self.act_buf,
            ret=self.ret_buf,
            adv=self.adv_buf,
            logp=self.logp_buf,
        )
        return data

    @staticmethod
    def merge(buf1: Dict[str, Any], buf2: Dict[str, Any]) -> Dict[str, torch.Tensor]:
        data_dict = {}
        for k, (v1, v2) in zip(buf1.keys(), zip(buf1.values(), buf2.values())):
            if isinstance(v1, torch.Tensor):
                v = torch.cat([v1, v2], dim=0)
            elif isinstance(v1, np.ndarray):
                v = np.concatenate([v1, v2], axis=0)
            else:
                raise TypeError
            data_dict[k] = v
        return data_dict


class Core:
    def __init__(
        self,
        actor_critic: ActorCritic,
        steps_per_epoch: int = 4000,
        epochs: int = 50,
        clip_ratio: float = 0.2,
        pi_lr: float = 3e-4,
        vf_lr: float = 1e-3,
        train_pi_iters: int = 80,
        train_v_iters: int = 80,
        max_ep_len: int = 1000,
        target_kl: float = 0.01,
        save_freq: int = 10,
        mini_batch: int = 500,
        use_norm_grad: bool = False,
        lr_decay: str = 'none',
    ) -> None:
        self._ac = actor_critic
        self._steps_per_epoch = steps_per_epoch
        self._epochs = epochs
        self._clip_ratio = clip_ratio
        self._train_pi_iters = train_pi_iters
        self._train_v_iters = train_v_iters
        self._max_ep_len = max_ep_len
        self._target_kl = target_kl
        self._save_freq = save_freq
        self._mini_batch = mini_batch
        self._pi_optimizer = Adam(self._ac.pi.parameters(), lr=pi_lr, eps=1e-5)
        self._vf_optimizer = Adam(self._ac.v.parameters(), lr=vf_lr, eps=1e-5)
        self._use_norm_grad = use_norm_grad
        if lr_decay == 'cosine':
            self._pi_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
                self._pi_optimizer, self._epochs
            )
            self._vf_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
                self._vf_optimizer, self._epochs
            )
        elif lr_decay == 'linear':
            self._pi_scheduler = torch.optim.lr_scheduler.LambdaLR(
                self._pi_optimizer, lambda e: 1 - e / self._epochs
            )
            self._vf_scheduler = torch.optim.lr_scheduler.LambdaLR(
                self._vf_optimizer, lambda e: 1 - e / self._epochs
            )
        else:
            self._pi_scheduler = None
            self._vf_scheduler = None
        self._device = get_available_device()
        self._ac.apply(weight_initialize)
        assert mini_batch <= steps_per_epoch

    @Logger.run_time("Core._process_data被调用")
    def _process_data(
        self, obs: Any, act: Any, logp_old: np.ndarray, adv: np.ndarray, ret: np.ndarray
    ) -> Tuple[Any, Any, torch.Tensor, torch.Tensor]:
        obs = self._ac.process_obs_batch(obs)
        if type(obs) == torch.Tensor:
            obs = obs.to(self._device)
        elif type(obs) == tuple:
            obs = tuple([o.to(self._device) for o in obs])  
        act = self._ac.process_act_batch(act).to(self._device)
        logp_old = torch.as_tensor(logp_old, dtype=torch.float32).to(self._device)
        adv = torch.as_tensor(adv, dtype=torch.float32).to(self._device)
        ret = torch.as_tensor(ret, dtype=torch.float32).to(self._device)
        return obs, act, logp_old, adv, ret

    @Logger.run_time("Core._compute_loss_pi被调用")
    def _compute_loss_pi(
        self, data: Dict[str, torch.Tensor]
    ) -> Tuple[torch.Tensor, Dict]:
        ent_list = []
        logp_list = []
        logp_old_list = []
        adv_list = []
        for obs, act, logp_old, adv, _ in self._data_iter(data):
            pi, logp = self.ac.pi(obs, act)
            ent_list.append(pi.entropy())
            logp_list.append(logp)
            logp_old_list.append(logp_old)
            adv_list.append(adv)
        ent = torch.cat(ent_list, dim=0)
        logp = torch.cat(logp_list, dim=0)
        logp_old = torch.cat(logp_old_list, dim=0)
        adv = torch.cat(adv_list, dim=0)

        ratio = torch.exp(logp - logp_old)
        clip_adv = torch.clamp(ratio, 1 - self._clip_ratio, 1 + self._clip_ratio) * adv
        loss_pi = -(torch.min(ratio * adv, clip_adv)).mean()

        # Useful extra info
        approx_kl = (logp_old - logp).mean().item()
        ent = ent.mean().item()
        clipped = ratio.gt(1 + self._clip_ratio) | ratio.lt(1 - self._clip_ratio)
        clipfrac = torch.as_tensor(clipped, dtype=torch.float32).mean().item()
        pi_info = dict(kl=approx_kl, ent=ent, cf=clipfrac)

        return loss_pi, pi_info

    @Logger.run_time("Core._compute_loss_v被调用")
    def _compute_loss_v(self, data: Dict[str, torch.Tensor]) -> torch.Tensor:
        val_list = []
        ret_list = []
        for obs, _, _, _, ret in self._data_iter(data):
            val_list.append(self.ac.v(obs))
            ret_list.append(ret)
        val = torch.cat(val_list, dim=0)
        ret = torch.cat(ret_list, dim=0)

        return ((val - ret) ** 2).mean()

    def _collect(self) -> Dict[str, torch.Tensor]:
        raise NotImplementedError

    def _data_iter(self, data: Dict[str, torch.Tensor]) -> Generator:
        for _ in range(data["obs"].shape[0] // self._mini_batch):
            rand_ids = np.random.choice(self._steps_per_epoch, self._mini_batch)
            obs, act, logp_old, adv, ret = (
                data["obs"][rand_ids],
                data["act"][rand_ids],
                data["logp"][rand_ids],
                data["adv"][rand_ids],
                data["ret"][rand_ids],
            )
            yield self._process_data(obs, act, logp_old, adv, ret)

    @Logger.run_time("Core._update被调用")
    def _update(
        self, data: Dict[str, torch.Tensor]
    ) -> Tuple[float, float, torch.Tensor, torch.Tensor, torch.Tensor]:
        update_flag = False
        for _ in range(self._train_pi_iters):
            self._pi_optimizer.zero_grad()
            loss_pi, pi_info = self._compute_loss_pi(data)
            kl = pi_info['kl']
            if kl > 1.5 * self._target_kl:
                break
            loss_pi.backward()
            if self._use_norm_grad:
                nn.utils.clip_grad_norm_(self.ac.pi.parameters(), 0.5)
            self._pi_optimizer.step()
            update_flag = True
        if self._pi_scheduler is not None and update_flag:
            self._pi_scheduler.step()

        update_flag = False
        for _ in range(self._train_v_iters):
            self._vf_optimizer.zero_grad()
            loss_v = self._compute_loss_v(data)
            loss_v.backward()
            if self._use_norm_grad:
                nn.utils.clip_grad_norm_(self.ac.v.parameters(), 0.5)
            self._vf_optimizer.step()
            update_flag = True
        if self._vf_scheduler is not None and update_flag:
            self._vf_scheduler.step()

        kl, ent, cf = pi_info['kl'], pi_info['ent'], pi_info['cf']
        return loss_pi.cpu().item(), loss_v.cpu().item(), kl, ent, cf

    def _eval(self) -> float:
        raise NotImplementedError

    @Logger.run_time("Core.train被调用")
    def train(self) -> None:
        """训练函数"""
        Logger.info("开始训练，使用的算法为PPO")
        Logger.log_num_of_parameters(self._ac.pi, "策略函数网络")
        Logger.log_num_of_parameters(self._ac.v, "值函数网络")
        writer = SummaryWriter(log_dir=Logger.ROOT)
        loss_pi_list = []
        loss_v_list = []
        reward_list = []
        kl_list = []
        entropy_list = []
        clipfrac_list = []
        for epoch in range(self._epochs):
            # 收集数据
            self._ac.to(torch.device("cpu"))
            data = self._collect()

            # 更新网络
            self._ac.train()
            self._ac.to(self._device)
            loss_pi, loss_v, kl, ent, cf = self._update(data)
            # 释放内存
            del data
            if self._device == torch.device("cuda"):
                torch.cuda.empty_cache()

            loss_pi_list.append(loss_pi)
            loss_v_list.append(loss_v)
            kl_list.append(kl)
            entropy_list.append(ent)
            clipfrac_list.append(cf)

            # 保存模型
            if (epoch % self._save_freq == 0) or (epoch == self._epochs - 1):
                Logger.save_data("pi", "pth", self._ac.pi.state_dict())
                Logger.save_data("v", "pth", self._ac.v.state_dict())

                self._ac.eval()
                self._ac.to(torch.device("cpu"))
                reward = self._eval()
                reward_list.append(reward)
                if isinstance(reward, float):
                    Logger.tb_add_scalar(writer, "scalars/reward", reward, epoch)
                else:
                    reward_dict = {"mean": reward.mean()}
                    for i, r in enumerate(reward):
                        reward_dict[str(i)] = r
                    Logger.tb_add_scalars(writer, "scalars/reward", reward_dict, epoch)

            Logger.tb_add_scalar(writer, "scalars/loss_pi", loss_pi, epoch)
            Logger.tb_add_scalar(writer, "scalars/loss_v", loss_v, epoch)
            Logger.tb_add_scalar(writer, "scalars/kl", kl, epoch)
            Logger.tb_add_scalar(writer, "scalars/entropy", ent, epoch)
            Logger.tb_add_scalar(writer, "scalars/clipfrac", cf, epoch)
            Logger.info(f"第{epoch+1}轮训练结束。")

        Logger.save_data("loss_pi", "npy", np.array(loss_pi_list))
        Logger.save_data("loss_v", "npy", np.array(loss_v_list))
        Logger.save_data("reward_list", "npy", np.array(reward_list))
        Logger.save_data("kl_list", "npy", np.array(kl_list))
        Logger.save_data("entropy_list", "npy", np.array(entropy_list))
        Logger.save_data("clipfrac_list", "npy", np.array(clipfrac_list))
        Logger.info(f"训练结束。")

    @property
    def ac(self) -> ActorCritic:
        return self._ac


def discount_cumsum(x, discount):
    """
    magic from rllab for computing discounted cumulative sums of vectors.

    input:
        vector x,
        [x0,
        x1,
        x2]

    output:
        [x0 + discount * x1 + discount^2 * x2,
        x1 + discount * x2,
        x2]
    """
    return lfilter([1], [1, float(-discount)], x[::-1], axis=0)[::-1]
